package newStreaming

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object window extends App {
  val sparkConf = new SparkConf()
  sparkConf.setAppName("wordCountDS")
  sparkConf.setMaster("local[4]")
  val sparkContext = new SparkContext(sparkConf)
  val ssc = new StreamingContext(sparkContext, Durations.seconds(10))
  ssc.checkpoint("data/windowSlide")
  // 先去node1 启动 nc -lk 9999
  val ds = ssc.socketTextStream("node2", 9999)

  val func = (x: Int, y: Int) => x + y

  val countDS = ds.flatMap(_.split(",")).map((_, 1))
    .reduceByKeyAndWindow(func, Durations.seconds(30), Durations.seconds(10))
  countDS.print()

  ssc.start()
  ssc.awaitTermination()
  ssc.stop()
}
